#define _XOPEN_SOURCE 700
#pragma execution_character_set("utf-8")
#include <time.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mysql/mysql.h>
#include <pthread.h>
#include "MQTTClient.h"

/*
    添加环境阈值下发嵌入式设备功能

*/

//需自己更改
#define ADDRESS     "tcp://13.110.112.231:1883"  // MQTT Broker 地址
#define CLIENTID    "MQTTToMySQLClient"
#define TOPIC       "kita/#"                     // 订阅kita/下的所有主题
#define TOPIC1       "status/isConnect/#"         // 订阅status/isConnect/下的所有主题
#define QOS         1
#define TIMEOUT     10000L

#define THRESHOLD_TOPIC_PREFIX "kita-thresholds/"  //下发阈值topic

#define DB_HOST     "47.120.16.231"
#define DB_USER     "root"
#define DB_PASS     "a2034318325"
#define DB_NAME     "mqtt"
#define MAX_SQL_LEN 1024
#define MAX_BATCH_SIZE 100

// 无效数据的定义
#define INVALID_TEMP -999
#define INVALID_HUMP -1
#define INVALID_LIGHT -1
#define INVALID_CO2 -1

#define BATCH_SIZE 50  // 每批插入的最大数据量

// 全局变量：环境数据和设备状态的缓存数组及其计数
char sql_buffer_environment[MAX_BATCH_SIZE][MAX_SQL_LEN];  // 用于缓存环境数据的SQL语句
char sql_buffer_status[MAX_BATCH_SIZE][MAX_SQL_LEN];       // 用于缓存设备状态的SQL语句
int sql_count_environment = 0;  // 当前环境数据的数量
int sql_count_status = 0;       // 当前设备状态的数量

//延时变量
struct timespec ts;

// 全局 MySQL 连接对象
MYSQL *global_conn = NULL;
// 用于保护 global_conn 的互斥锁
pthread_mutex_t mysql_lock = PTHREAD_MUTEX_INITIALIZER;
//防止并发读写冲突互斥锁
pthread_mutex_t mutex_threshold_list = PTHREAD_MUTEX_INITIALIZER;


pthread_mutex_t mutex_environment = PTHREAD_MUTEX_INITIALIZER;  // 用于保护环境数据缓存
pthread_mutex_t mutex_status = PTHREAD_MUTEX_INITIALIZER;       // 用于保护设备状态缓存

// 用户阈值链表节点结构体
typedef struct UserThreshold {
    int user_id;  // 用户ID
    int temp_sensor_id;  // 温湿度传感器的设备ID
    int co2_sensor_id;  // 气体传感器的设备ID
    int light_sensor_id;  // 光照传感器的设备ID
    float temp_min;
    float temp_max;
    float humidity_min;
    float humidity_max;
    int light_min;
    int light_max;
    int co2_max;
    struct UserThreshold* next;  // 指向下一个节点
} UserThreshold;

// 创建用户阈值链表头
UserThreshold* head = NULL;  // 链表头

//声明
void insert_user_threshold(UserThreshold* new_node);

MYSQL* connect_to_mysql() {
    return global_conn;
}

int safe_mysql_query(const char *query) {
    pthread_mutex_lock(&mysql_lock);
    int res = mysql_query(global_conn, query);
    pthread_mutex_unlock(&mysql_lock);
    return res;
}

int safe_mysql_queryf(const char *fmt, ...) {
    char query[1024];
    va_list args;
    va_start(args, fmt);
    vsnprintf(query, sizeof(query), fmt, args);
    va_end(args);

    return safe_mysql_query(query);
}

MYSQL_RES* safe_mysql_store_result() {
    pthread_mutex_lock(&mysql_lock);
    MYSQL_RES* res = mysql_store_result(global_conn);
    pthread_mutex_unlock(&mysql_lock);
    return res;
}

void insert_user_threshold_safe(UserThreshold* new_node) {
    pthread_mutex_lock(&mutex_threshold_list);
    insert_user_threshold(new_node);  // 原始版本的插入函数
    pthread_mutex_unlock(&mutex_threshold_list);
}

void clear_threshold_list() {
    pthread_mutex_lock(&mutex_threshold_list);
    UserThreshold *curr = head;
    while (curr) {
        UserThreshold *next = curr->next;
        free(curr);
        curr = next;
    }
    head = NULL;
    pthread_mutex_unlock(&mutex_threshold_list);
}

int init_mysql_once() {
    global_conn = mysql_init(NULL);
    if (!global_conn) {
        fprintf(stderr, "mysql_init() failed\n");
        return 0;
    }

    if (!mysql_real_connect(global_conn, DB_HOST, DB_USER, DB_PASS, DB_NAME, 0, NULL, 0)) {
        fprintf(stderr, "mysql_real_connect() failed: %s\n", mysql_error(global_conn));
        global_conn = NULL;
        return 0;
    }

    if (mysql_set_character_set(global_conn, "utf8mb4")) {
        fprintf(stderr, "mysql_set_character_set() failed: %s\n", mysql_error(global_conn));
    }

    return 1;
}


// 从kita主题中提取用户ID
void extract_user_id_from_topic(const char *topic, int *user_id) {
    if (sscanf(topic, "kita/%d", user_id) != 1) {
        *user_id = -1;  // 如果主题格式不正确，返回 -1
    }
}

// 从status/isConnect主题中提取用户ID
void extract_user_id_from_topic_status(const char *topic, int *user_id) {
    if (sscanf(topic, "status/isConnect/%d", user_id) != 1) {
        *user_id = -1;  // 如果主题格式不正确，返回 -1
    }
}

// 插入设备状态到缓存
void insert_device_status_to_buffer(int device_id, const char* status) {
    char value[256];

    // 构建设备状态 SQL 语句
    snprintf(value, sizeof(value), "(%d, '%s', NOW())", device_id, status);

    pthread_mutex_lock(&mutex_status);

    // 将设备状态数据插入缓存
    if (sql_count_status < MAX_BATCH_SIZE) {
        strncpy(sql_buffer_status[sql_count_status], value, MAX_SQL_LEN);
        sql_count_status++;
    }

    pthread_mutex_unlock(&mutex_status);
}

//插入报警信息的函数
void insert_alert(MYSQL *conn, int device_id, const char* alert_type, const char* message) {
    char check_query[512];
    snprintf(check_query, sizeof(check_query),
             "SELECT triggered_at FROM alerts WHERE device_id = %d AND alert_type = '%s' ORDER BY triggered_at DESC LIMIT 1",
             device_id, alert_type);

    if (safe_mysql_query(check_query)) {
        fprintf(stderr, "Check alert query failed: %s\n", mysql_error(global_conn));
        return;
    }

    MYSQL_RES *res = safe_mysql_store_result();
    if (res == NULL) {
        fprintf(stderr, "Check result failed: %s\n", mysql_error(global_conn));
        return;
    }

    int allow_insert = 1;  // 是否允许插入报警
    MYSQL_ROW row = mysql_fetch_row(res);
    if (row) {
        struct tm last_tm = {0};
        strptime(row[0], "%Y-%m-%d %H:%M:%S", &last_tm);
        time_t last_time = mktime(&last_tm);
        time_t now = time(NULL);

        if (difftime(now, last_time) < 600) {  // 600秒内不重复报警
            allow_insert = 0;
        }
    }
    mysql_free_result(res);

    if (!allow_insert) {
        printf("Skipping duplicate alert (within 10 mins): %s - %s\n", alert_type, message);
        return;
    }

    // 插入报警
    char query[512];
    snprintf(query, sizeof(query),
             "INSERT INTO alerts (device_id, alert_type, message, triggered_at) "
             "VALUES (%d, '%s', '%s', NOW())",
             device_id, alert_type, message);

    if (safe_mysql_query(query)) {
        fprintf(stderr, "Alert insert failed: %s\n", mysql_error(global_conn));
    } else {
        printf("Alert inserted: [%d] %s - %s\n", device_id, alert_type, message);
    }
}


// 插入传感器数据到缓存
void insert_data_to_buffer(int device_id, int sensor_type, float temp, float humidity, int light, int co2) {
    char value[256];

    // 构建SQL语句
    if (sensor_type == 1) {
        snprintf(value, sizeof(value), "(%d, %d, NOW(), NULL, NULL, NULL, %d)", device_id, sensor_type, co2);
    } else if (sensor_type == 2) {
        snprintf(value, sizeof(value), "(%d, %d, NOW(), NULL, NULL, %d, NULL)", device_id, sensor_type, light);
    } else if (sensor_type == 3) {
        snprintf(value, sizeof(value), "(%d, %d, NOW(), %f, %f, NULL, NULL)", device_id, sensor_type, temp, humidity);
    }

    pthread_mutex_lock(&mutex_environment);

    // 将环境数据插入缓存
    if (sql_count_environment < MAX_BATCH_SIZE) {
        strncpy(sql_buffer_environment[sql_count_environment], value, MAX_SQL_LEN);
        sql_count_environment++;
    }

    pthread_mutex_unlock(&mutex_environment);
}

// 打印所有用户阈值链表
void print_user_thresholds() {
    pthread_mutex_lock(&mutex_threshold_list);
    UserThreshold* current = head;
    while (current != NULL) {
        printf("User ID: %d\n", current->user_id);
        printf("Temp Sensor ID: %d\n", current->temp_sensor_id);
        printf("CO2 Sensor ID: %d\n", current->co2_sensor_id);
        printf("Light Sensor ID: %d\n", current->light_sensor_id);
        printf("Temp Min: %.2f, Temp Max: %.2f\n", current->temp_min, current->temp_max);
        printf("Humidity Min: %.2f, Humidity Max: %.2f\n", current->humidity_min, current->humidity_max);
        printf("Light Min: %d, Light Max: %d\n", current->light_min, current->light_max);
        printf("CO2 Max: %d\n", current->co2_max);
        printf("\n");
        current = current->next;
    }
    pthread_mutex_unlock(&mutex_threshold_list);
}

// 批量插入函数（插入到数据库）
// data_type = 1 -> 环境数据插入到 environment_data 表
// data_type = 2 -> 设备状态插入到 device_status_log 表
void batch_insert_data(MYSQL *conn, int data_type) {
    char query[4096];

    if (data_type == 1) {  // 环境数据插入
        snprintf(query, sizeof(query), "INSERT INTO environment_data (device_id, sensor_type_id, timestamp, temperature, humidity, light, co2) VALUES ");
        for (int i = 0; i < sql_count_environment; i++) {
            strcat(query, sql_buffer_environment[i]);
            if (i < sql_count_environment - 1) {
                strcat(query, ", ");
            }
        }
    } else if (data_type == 2) {  // 设备状态插入
        snprintf(query, sizeof(query), "INSERT INTO device_status_log (device_id, status, timestamp) VALUES ");
        for (int i = 0; i < sql_count_status; i++) {
            strcat(query, sql_buffer_status[i]);
            if (i < sql_count_status - 1) {
                strcat(query, ", ");
            }
        }
    }

    // 执行插入操作
    if (safe_mysql_query(query)) {
        fprintf(stderr, "INSERT failed: %s\n", mysql_error(conn));
    } else {
        printf("Batch data inserted successfully.\n");
    }

    // 清空缓存
    if (data_type == 1) {
        sql_count_environment = 0;
    } else if (data_type == 2) {
        sql_count_status = 0;
    }
}

// 获取设备的ID
void get_device_ids(MYSQL *conn, int user_id, int *device_ids, int *sensor_types) {
    char query[256];
    snprintf(query, sizeof(query), "SELECT id, sensor_type_id FROM devices WHERE user_id = %d", user_id);

    if (safe_mysql_query(query)) {
        fprintf(stderr, "Query failed: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_RES *res = safe_mysql_store_result();
    if (res == NULL) {
        fprintf(stderr, "Failed to fetch result: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_ROW row;
    int i = 0;
    while ((row = mysql_fetch_row(res))) {
        device_ids[i] = atoi(row[0]);
        sensor_types[i] = atoi(row[1]);
        i++;
    }
    //调试
    printf("\nsensor_types[0] : %d\n", sensor_types[0]);
    printf("\nsensor_types[1] : %d\n", sensor_types[1]);
    printf("\nsensor_types[2] : %d\n", sensor_types[2]);
    mysql_free_result(res);
}

// 获取用户设备ID（包括温湿度、气体、光照传感器设备ID）
void get_device_ids_sensor(MYSQL *conn, int user_id, int *temp_sensor_id, int *co2_sensor_id, int *light_sensor_id) {
    char query[256];
    snprintf(query, sizeof(query), "SELECT id, sensor_type_id FROM devices WHERE user_id = %d", user_id);

    if (safe_mysql_query(query)) {
        fprintf(stderr, "Query failed: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_RES *res = safe_mysql_store_result();
    if (res == NULL) {
        fprintf(stderr, "Failed to fetch result: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_ROW row;
    while ((row = mysql_fetch_row(res))) {
        int sensor_type_id = atoi(row[1]);
        int device_id = atoi(row[0]);

        // 根据传感器类型ID分配设备ID
        if (sensor_type_id == 1) {
            *temp_sensor_id = device_id; // 温湿度传感器
        } else if (sensor_type_id == 2) {
            *light_sensor_id = device_id; // 光照传感器
        } else if (sensor_type_id == 3) {
            *co2_sensor_id = device_id; // 气体传感器
        }
    }
    mysql_free_result(res);
}

//发送 JSON 阈值数据
void publish_threshold_to_user(MQTTClient *client, UserThreshold* user) {
    char topic[64];
    snprintf(topic, sizeof(topic), THRESHOLD_TOPIC_PREFIX "%d", user->user_id);

    char payload[512];
    snprintf(payload, sizeof(payload),
             "{ temp_min: %.2f, temp_max: %.2f, humidity_min: %.2f, humidity_max: %.2f, light_min: %d, light_max: %d, co2_max: %d }",
             user->temp_min, user->temp_max,
             user->humidity_min, user->humidity_max,
             user->light_min, user->light_max,
             user->co2_max);

    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    pubmsg.payload = payload;
    pubmsg.payloadlen = strlen(payload);
    pubmsg.qos = 1;
    pubmsg.retained = 0;

    MQTTClient_deliveryToken token;
    int rc = MQTTClient_publishMessage(*client, topic, &pubmsg, &token);
    if (rc != MQTTCLIENT_SUCCESS) {
        printf("Failed to publish threshold to user %d\n", user->user_id);
    } else {
        MQTTClient_waitForCompletion(*client, token, 10000L);
        printf("Threshold published to user %d: %s\n", user->user_id, payload);
    }
}

// 遍历整个链表，下发所有用户阈值
void publish_all_thresholds(MQTTClient *client, UserThreshold* head) {
    UserThreshold* current = head;
    while (current) {
        publish_threshold_to_user(client, current);
        current = current->next;
    }
}

// 获取传感器阈值
void get_sensor_threshold(MYSQL *conn, int user_id, int sensor_id, int sensor_type_id, UserThreshold *user_threshold) {
    char query[512];

    if (sensor_type_id == 3) {  // 获取温湿度传感器的阈值
        snprintf(query, sizeof(query),
                 "SELECT temp_min, temp_max, humidity_min, humidity_max FROM user_settings "
                 "WHERE user_id = %d AND sensor_type_id = 3 ORDER BY id DESC LIMIT 1",
                 user_id);
    } else if (sensor_type_id == 2) {  // 获取光照传感器的阈值
        snprintf(query, sizeof(query),
                 "SELECT light_min, light_max FROM user_settings "
                 "WHERE user_id = %d AND sensor_type_id = 2 ORDER BY id DESC LIMIT 1",
                 user_id);
    } else if (sensor_type_id == 1) {  // 获取气体传感器的阈值
        snprintf(query, sizeof(query),
                 "SELECT co2_max FROM user_settings "
                 "WHERE user_id = %d AND sensor_type_id = 1 ORDER BY id DESC LIMIT 1",
                 user_id);
    }

    if (safe_mysql_query(query)) {
        fprintf(stderr, "Query failed: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_RES *res = safe_mysql_store_result();
    if (res == NULL) {
        fprintf(stderr, "Failed to fetch result: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_ROW row;
    if ((row = mysql_fetch_row(res))) {
        if (sensor_type_id == 3) {  // 温湿度传感器
            user_threshold->temp_min = atof(row[0] ? row[0] : "0");
            user_threshold->temp_max = atof(row[1] ? row[1] : "0");
            user_threshold->humidity_min = atof(row[2] ? row[2] : "0");
            user_threshold->humidity_max = atof(row[3] ? row[3] : "0");
        } else if (sensor_type_id == 2) {  // 光照传感器
            user_threshold->light_min = atoi(row[0] ? row[0] : "0");
            user_threshold->light_max = atoi(row[1] ? row[1] : "0");
        } else if (sensor_type_id == 1) {  // 气体传感器
            user_threshold->co2_max = atoi(row[0] ? row[0] : "0");
        }
    }

    mysql_free_result(res);
}

// 插入用户阈值到链表
void insert_user_threshold(UserThreshold* new_node) {
    if (head == NULL || head->user_id > new_node->user_id) {
        new_node->next = head;
        head = new_node;
    } else {
        UserThreshold* current = head;
        while (current->next != NULL && current->next->user_id < new_node->user_id) {
            current = current->next;
        }
        new_node->next = current->next;
        current->next = new_node;
    }
}

// 获取并存储所有用户的设备ID及阈值
void load_user_thresholds_from_db(MYSQL *conn) {
    char query[512];
    snprintf(query, sizeof(query), "SELECT DISTINCT user_id FROM user_settings");

    if (safe_mysql_query(query)) {
        fprintf(stderr, "Query failed: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_RES *res = safe_mysql_store_result();
    if (res == NULL) {
        fprintf(stderr, "Failed to fetch result: %s\n", mysql_error(conn));
        return;
    }

    MYSQL_ROW row_users;
    while ((row_users = mysql_fetch_row(res))) {
        int user_id = atoi(row_users[0]);

        // 获取该用户绑定的设备ID
        int temp_sensor_id = 0, co2_sensor_id = 0, light_sensor_id = 0;
        get_device_ids_sensor(conn, user_id, &temp_sensor_id, &co2_sensor_id, &light_sensor_id);

        // 获取该用户各个传感器的阈值
        UserThreshold* user_threshold = (UserThreshold*)malloc(sizeof(UserThreshold));
        user_threshold->user_id = user_id;
        user_threshold->temp_sensor_id = temp_sensor_id;
        user_threshold->co2_sensor_id = co2_sensor_id;
        user_threshold->light_sensor_id = light_sensor_id;

        // 获取温湿度传感器阈值
        get_sensor_threshold(conn, user_id, temp_sensor_id, 3, user_threshold);
        // 获取光照传感器阈值
        get_sensor_threshold(conn, user_id, light_sensor_id, 2, user_threshold);
        // 获取气体传感器阈值
        get_sensor_threshold(conn, user_id, co2_sensor_id, 1, user_threshold);

        // 将获取到的信息存入链表
        user_threshold->next = NULL;
        insert_user_threshold_safe(user_threshold);
    }

    mysql_free_result(res);
}

//后台线程：每天 00:00 的下发阈值线程逻辑
void* daily_threshold_publish_thread(void* arg) {
    MQTTClient *client = (MQTTClient *)arg;

    while (1) {
        time_t now = time(NULL);
        struct tm *tm_now = localtime(&now);

        tm_now->tm_hour = 0;
        tm_now->tm_min = 0;
        tm_now->tm_sec = 0;

        // time_t next_run = mktime(tm_now) + 86400; // 下一天0点
        // sleep(next_run - now); // 睡到明天

        sleep(30);

        if (global_conn) {
            clear_threshold_list();

            load_user_thresholds_from_db(global_conn ); // 重新加载最新阈值
            publish_all_thresholds(client, head); // 发布
        }
    }
    return NULL;
}


//后台线程: 周期性刷新阈值
void* threshold_refresh_thread(void* arg) {
    while (1) {
        if (global_conn) {
            clear_threshold_list();

            // 重新加载
            load_user_thresholds_from_db(global_conn );

        }
        sleep(3000);  // 每 50 分钟刷新一次
    }
    return NULL;
}

// 后台线程：实时插入
void* real_time_insert_thread(void* arg) {
    if (global_conn == NULL) {
        printf("Failed to connect to MySQL\n");
        return NULL;
    }

    while (1) {
        pthread_mutex_lock(&mutex_environment); // 使用环境数据的互斥锁

        // 判断缓存中是否有数据，并根据数据类型执行插入
        if (sql_count_environment > 0) {
            // 执行环境数据批量插入
            batch_insert_data(global_conn , 1);  // 1 表示插入环境数据
        }

        pthread_mutex_unlock(&mutex_environment);

        pthread_mutex_lock(&mutex_status);  // 使用设备状态缓存的互斥锁

        if (sql_count_status > 0) {
            // 执行设备状态批量插入
            batch_insert_data(global_conn , 2);  // 2 表示插入设备状态
        }

        pthread_mutex_unlock(&mutex_status);  // 解锁

        // 等待一小段时间后再次检查缓存
        ts.tv_sec = 0;
        ts.tv_nsec = 10000000;  // 10毫秒 = 10,000,000 纳秒
        nanosleep(&ts, NULL);
    }

    return NULL;
}

// 设备上线和下线处理
void handle_device_status(MYSQL *conn, int user_id, const char* status) {
    // 查询该用户绑定的传感器设备ID
    int device_ids[3], sensor_types[3];
    get_device_ids(conn, user_id, device_ids, sensor_types);

    // 根据状态更新所有传感器设备的状态
    for (int i = 0; i < 3; i++) {
        // 将设备上线或下线的状态插入到缓存中
        insert_device_status_to_buffer(device_ids[i], status);
    }
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    printf("Message arrived on topic: %s\n", topicName);
    printf("Message: %.*s\n", message->payloadlen, (char*)message->payload);

    int user_id = -1;
    const char* valid_topic1 = "status/isConnect/";

    // 设备上线/下线处理
    if (strncmp(topicName, valid_topic1, strlen(valid_topic1)) == 0) {
        extract_user_id_from_topic_status(topicName, &user_id);
        if (user_id == -1) {
            printf("Invalid topic format: %s\n", topicName);
            return 1;
        }

        if (global_conn) {
            if (strcmp((char*)message->payload, "clientID is online") == 0) {
                handle_device_status(global_conn, user_id, "online");
            } else if (strcmp((char*)message->payload, "clientID is offline") == 0) {
                handle_device_status(global_conn, user_id, "offline");
            }
        }
        MQTTClient_freeMessage(&message);
        MQTTClient_free(topicName);
        return 1;
    }

    // 处理 kita 主题
    const char* valid_topic = "kita/";
    if (strncmp(topicName, valid_topic, strlen(valid_topic)) == 0) {
        extract_user_id_from_topic(topicName, &user_id);

        if (user_id == -1) {
            printf("Invalid topic format: %s\n", topicName);
            return 1;
        }

        float temp, humidity;
        int light, co2;
        int num = sscanf(message->payload, "{Temp_data:%f,Hump_data:%f,GY_data:%04d,CO2Data:%04d}",
                         &temp, &humidity, &light, &co2);

        if (num != 4) {
            printf("Failed to parse payload: %s\n", (char*)message->payload);
            return 1;
        }

        int device_ids[3], sensor_types[3];
        get_device_ids(connect_to_mysql(), user_id, device_ids, sensor_types);
        for (int i = 0; i < 3; i++) {
            insert_data_to_buffer(device_ids[i], sensor_types[i], temp, humidity, light, co2);
        }

        if (global_conn) {
            pthread_mutex_lock(&mutex_threshold_list);
            UserThreshold* current = head;
            while (current) {
                if (current->user_id == user_id) {
                    // 温度判断
                    if (temp < current->temp_min) {
                        char msg[128];
                        //snprintf(msg, sizeof(msg), "Temperature is below the threshold %.2f°C, current value is %.2f°C", current->temp_min, temp);
                        snprintf(msg, sizeof(msg), "温度低于设定值 %.2f°C，当前值为 %.2f°C", current->temp_min, temp);
                        insert_alert(global_conn, current->temp_sensor_id, "TEMP_LOW", msg);
                    } else if (temp > current->temp_max) {
                        char msg[128];
                        //snprintf(msg, sizeof(msg), "Temperature is above the threshold %.2f°C, current value is %.2f°C", current->temp_max, temp);
                        snprintf(msg, sizeof(msg), "温度高于设定值 %.2f°C，当前值为 %.2f°C", current->temp_max, temp);
                        insert_alert(global_conn, current->temp_sensor_id, "TEMP_HIGH", msg);
                    }

                    // 湿度判断
                    if (humidity < current->humidity_min) {
                        char msg[128];
                        //snprintf(msg, sizeof(msg), "Humidity is below the threshold %.2f%%, current value is %.2f%%", current->humidity_min, humidity);
                        snprintf(msg, sizeof(msg), "湿度低于设定值 %.2f%%，当前值为 %.2f%%", current->humidity_min, humidity);
                        insert_alert(global_conn, current->temp_sensor_id, "HUMP_LOW", msg);
                    } else if (humidity > current->humidity_max) {
                        char msg[128];
                        //snprintf(msg, sizeof(msg), "Humidity is above the threshold %.2f%%, current value is %.2f%%", current->humidity_max, humidity);
                        snprintf(msg, sizeof(msg), "湿度高于设定值 %.2f%%，当前值为 %.2f%%", current->humidity_max, humidity);
                        insert_alert(global_conn, current->temp_sensor_id, "HUMP_HIGH", msg);
                    }

                    // 光照判断
                    if (light < current->light_min) {
                        char msg[128];
                        //snprintf(msg, sizeof(msg), "Light intensity is below the threshold %d lx, current value is %d lx", current->light_min, light);
                        snprintf(msg, sizeof(msg), "光照低于设定值 %d lx，当前值为 %d lx", current->light_min, light);
                        insert_alert(global_conn, current->light_sensor_id, "LIGHT_LOW", msg);
                    } else if (light > current->light_max) {
                        char msg[128];
                        //snprintf(msg, sizeof(msg), "Light intensity is above the threshold %d lx, current value is %d lx", current->light_max, light);
                        snprintf(msg, sizeof(msg), "光照高于设定值 %d lx，当前值为 %d lx", current->light_max, light);
                        insert_alert(global_conn, current->light_sensor_id, "LIGHT_HIGH", msg);
                    }

                    // CO2 判断（仅上限）
                    if (co2 > current->co2_max) {
                        char msg[128];
                        //snprintf(msg, sizeof(msg), "CO? concentration exceeds the threshold %d ppm, current value is %d ppm", current->co2_max, co2);
                        snprintf(msg, sizeof(msg), "CO? 超过设定值 %d ppm，当前值为 %d ppm", current->co2_max, co2);
                        insert_alert(global_conn, current->co2_sensor_id, "CO2_HIGH", msg);
                    }

                    break;
                }
                current = current->next;
            }
            pthread_mutex_unlock(&mutex_threshold_list);

        }
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}


void connlost(void *context, char *cause) {
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int main(int argc, char *argv[]) {

    if (!init_mysql_once()) {
        return EXIT_FAILURE;
    }

    // 启动 获取并加载所有用户的阈值 线程
    pthread_t threshold_thread;
    pthread_create(&threshold_thread, NULL, threshold_refresh_thread, NULL);
    pthread_detach(threshold_thread);// 使线程在插入完成后自动回收


    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    // 创建MQTT客户端
    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    // 设置回调
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, NULL);

    // 连接MQTT服务器
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }


    //更新嵌入式设备阈值
    pthread_t threshold_publish_thread;
    pthread_create(&threshold_publish_thread, NULL, daily_threshold_publish_thread, &client);
    pthread_detach(threshold_publish_thread);

    // 启动实时插入线程
    pthread_t insert_thread;
    pthread_create(&insert_thread, NULL, real_time_insert_thread, NULL);
    pthread_detach(insert_thread);  // 使线程在插入完成后自动回收

    // 订阅kita/#主题
    MQTTClient_subscribe(client, TOPIC, QOS);
    // 订阅status/isConnect/#主题
    MQTTClient_subscribe(client, TOPIC1, QOS);

    // 等待消息
    while (1) {
      sleep(60);
      // 打印所有用户阈值
      print_user_thresholds();
        // 无限循环，等待消息到达
    }


    // 断开连接并销毁客户端
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);

    if (global_conn) mysql_close(global_conn);

    return rc;
}
